package com.hliushi.spark.rdd

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

import scala.collection.mutable

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/16 21:29
 */
class Accumulator {


  val conf: SparkConf = new SparkConf().setAppName("acc").setMaster("local[6]")
  val sc = new SparkContext(conf)

  /**
   * RDD -> ("a", "b", "c", "d") -> Set(c, d, a, b)
   */
  @Test
  def acc(): Unit = {
    val numAcc = new NumAccumulator()
    // 注册给Spark
    sc.register(numAcc, "num")

    sc.parallelize(Seq("a", "b", "c", "d"))
      .foreach((item: String) => numAcc.add(item))

    // numAcc.value = Set(c, d, a, b)
    println(s"numAcc.value = ${numAcc.value}")

    sc.stop()
  }
}


/**
 * 自定义累加器
 */
class NumAccumulator extends AccumulatorV2[String, Set[String]] {

  private val nums: mutable.Set[String] = mutable.Set()


  /**
   * 告诉Spark框架, 这个累加器对象是否为空的
   *
   * @return
   */
  override def isZero: Boolean = {
    nums.isEmpty
  }

  /**
   * 提供给Spark框架一个拷贝的累加器
   *
   * @return
   */
  override def copy(): AccumulatorV2[String, Set[String]] = {
    val newAccumulator = new NumAccumulator

    /**
     *  nums.synchronized {} 是什么? 2021年5月16日21:40:09
     */
    nums.synchronized {
      newAccumulator.nums ++= this.nums
    }
    newAccumulator
  }


  /**
   * 帮助Spark框架, 清理累加器的内容
   */
  override def reset(): Unit = {
    nums.clear()
  }

  /**
   * 外部传入要累加的内容, 在这个方法中进行累加
   *
   * @param v
   */
  override def add(v: String): Unit = {
    nums += v
  }

  /**
   * 累加器在进行累加的时候, 可能每个分布式节点都有一个实例
   * 在最后Driver进行一次合并, 把所有的实例的内容合并起来, 会调用这个merge方法进行合并
   *
   * @param other
   */
  override def merge(other: AccumulatorV2[String, Set[String]]): Unit = {
    nums ++= other.value
  }

  /**
   * 提供给外部结果
   * #
   * 为什么一定要给不可变的Set ???
   * .    因为外部有可能再进行修改, 如果是可变的集合, 其外部修改会影响内部的值
   *
   * @return
   */
  override def value: Set[String] = {
    nums.toSet
  }
}